package com.example.yckjbigdataflink.transformation;

import com.example.yckjbigdataflink.model.UserAction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 *
 * 调用示例
 *
 * DataStream<UserAction> input = ... // your HBase-derived stream
 *
 * DataStream<UserAction> latestStream = input
 *     .keyBy(UserAction::getUid)  // 以 uid 分组
 *     .process(new LatestUserActionProcessFunction());
 */
// 功能描述：每个用户只保留最新一次行为记录，我们就可以将这段处理提取成一个类。
public class LatestUserActionProcessFunction extends KeyedProcessFunction<String, UserAction, UserAction> {

    private transient ValueState<UserAction> latestState;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<UserAction> descriptor = new ValueStateDescriptor<>(
                "latest-user-action", UserAction.class);
        latestState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(UserAction value, Context ctx, Collector<UserAction> out) throws Exception {
        UserAction current = latestState.value();

        if (current == null || value.getEventTime() > current.getEventTime()) {
            latestState.update(value);
            out.collect(value); // emit only the latest // 输出最新
        }
    }
}
